"""
Case Type   : 权限
Case Name   : show grants--显示访问global setting和column setting的权限
Create At   : 2024.8.19
Owner       : wang-tianjing1
Description :
    1、创建兼容B库
    2、创建管理员用户
    3、分别创建cmk和cek
    4、授予权限并查询权限
    5、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
    5、成功
History     :
    modified by @xuwenfei1 2025-3.19:在执行show grants for user前，将参数b_compatibility_user_host_auth的值设置为true
"""

import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class OpreatorTest(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_primary = CommonSH('PrimaryDbBmode')
        self.user_node = Node('PrimaryDbBmode')
        self.constant = Constant()
        self.db_user = 'u_sg_tester09'
        self.db_name = 'grammar_show_db'
        self.tmp_path = os.path.join('$GAUSSHOME', 'etc',
                                     'localkms')
        status = self.sh_primary.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

    def test_show(self):
        text = '----step1:创建兼容B库;expect:成功----'
        self.log.info(text)
        sql_cmd = f'set dolphin.b_compatibility_mode=off;' \
                  f'drop database if exists {self.db_name};' \
                  f'create database {self.db_name} dbcompatibility \'B\';'
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertIn('DROP DATABASE', sql_res, '执行失败' + text)
        self.assertIn('CREATE DATABASE', sql_res, '执行失败' + text)

        text = '----step2:创建管理员用户;expect:成功----'
        self.log.info(text)
        sql_cmd = f'set current_schema to public;' \
                  f'show current_schema;' \
                  f"drop user if exists {self.db_user} cascade;" \
                  f"create user {self.db_user} with sysadmin password" \
                  f"'{macro.COMMON_PASSWD}';" \
                  f"grant all privileges to {self.db_user};" \
                  f'set session authorization {self.db_user} password' \
                  f"'{macro.COMMON_PASSWD}';"
        self.log.info(sql_cmd)
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('SET', sql_res, '执行失败' + text)
        self.assertIn('public', sql_res, '执行失败' + text)
        self.assertIn('DROP ROLE', sql_res, '执行失败' + text)
        self.assertIn('CREATE ROLE', sql_res, '执行失败' + text)
        self.assertIn('ALTER ROLE', sql_res, '执行失败' + text)
        self.assertIn('SET', sql_res, '执行失败' + text)

        text = '----step3:分别创建cmk和cek;expect:成功----'
        self.log.info(text)
        self.log.info("===创建localkms路径，except：成功===")
        mkdir_cmd = f'source {macro.DB_ENV_PATH};' \
                    f'rm -rf {self.tmp_path};' \
                    f'mkdir {self.tmp_path};' \
                    f'echo $GAUSSHOME' \
                    f'ls {self.tmp_path}'
        self.log.info(mkdir_cmd)
        mkdir_res = self.user_node.sh(mkdir_cmd).result()
        self.log.info(mkdir_res)
        self.assertIn("/app", mkdir_res, '执行失败' + text)
        self.assertIn("localkms", mkdir_res, '执行失败' + text)

        self.log.info("===创建域，except：成功===")
        sql_cmd2 = f'drop client master key if exists cmk_show cascade;' \
                   f'create client master key cmk_show with' \
                   f'(key_store = localkms, key_path = \"key_path_value\",' \
                   f'algorithm = rsa_2048);' \
                   f'drop column encryption key if exists cek_show cascade;' \
                   f'create column encryption key cek_show with values' \
                   f' (client_master_key = cmk_show, ' \
                   f'algorithm = aead_aes_256_cbc_hmac_sha256);'
        self.log.info(sql_cmd2)
        connect = f'-d {self.db_name} -U {self.db_user} -W {macro.COMMON_PASSWD} -C'
        sql_res2 = self.sh_primary.execut_db_sql(sql_cmd2, sql_type=connect)
        self.log.info(sql_res2)
        self.assertIn('DROP CLIENT MASTER KEY', sql_res2, '执行失败' + text)
        self.assertIn('CREATE CLIENT MASTER KEY', sql_res2, '执行失败' + text)
        self.assertIn('DROP COLUMN ENCRYPTION KEY', sql_res2, '执行失败' + text)
        self.assertIn('CREATE COLUMN ENCRYPTION KEY', sql_res2, '执行失败' + text)

        text = '----step4:授予权限,expect:成功----'
        self.log.info(text)
        sql_cmd3 = f'grant all privileges on column_encryption_key cek_show to {self.db_user};' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd3)
        sql_res3 = self.sh_primary.execut_db_sql(sql_cmd3, sql_type=connect)
        self.log.info(sql_res3)
        self.assertIn('GRANT', sql_res3, '执行失败' + text)
        self.assertIn('cmk_show', sql_res3, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res3, '执行失败' + text)

        sql_cmd4 = f'grant all privileges on client_master_key cmk_show to {self.db_user};' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd4)
        sql_res4 = self.sh_primary.execut_db_sql(sql_cmd4, sql_type=connect)
        self.log.info(sql_res4)
        self.assertIn('GRANT', sql_res4, '执行失败' + text)
        self.assertIn('cmk_show', sql_res4, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(5 rows)', sql_res4, '执行失败' + text)

        sql_cmd5 = f'grant usage on column_encryption_key cek_show to {self.db_user};' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd5)
        sql_res5 = self.sh_primary.execut_db_sql(sql_cmd5, sql_type=connect)
        self.log.info(sql_res5)
        self.assertIn('GRANT', sql_res5, '执行失败' + text)
        self.assertIn('cmk_show', sql_res5, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(5 rows)', sql_res5, '执行失败' + text)

        sql_cmd6 = f'grant usage on client_master_key cmk_show to {self.db_user};' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd6)
        sql_res6 = self.sh_primary.execut_db_sql(sql_cmd6, sql_type=connect)
        self.log.info(sql_res6)
        self.assertIn('GRANT', sql_res6, '执行失败' + text)
        self.assertIn('cmk_show', sql_res6, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(5 rows)', sql_res6, '执行失败' + text)

        sql_cmd7 = f'revoke all privileges on column_encryption_key cek_show from {self.db_user};' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd7)
        sql_res7 = self.sh_primary.execut_db_sql(sql_cmd7, sql_type=connect)
        self.log.info(sql_res7)
        self.assertIn('REVOKE', sql_res7, '执行失败' + text)
        self.assertIn('cmk_show', sql_res7, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res7, '执行失败' + text)

        sql_cmd8 = f'revoke usage on column_encryption_key cek_show from u_sg_tester09;' \
                   f'select * from gs_client_global_keys where global_key_name  = \'cmk_show\';' \
                   f'set b_compatibility_user_host_auth  = true;' \
                   f'show grants for \'{self.db_user}\';'
        self.log.info(sql_cmd8)
        sql_res8 = self.sh_primary.execut_db_sql(sql_cmd8, sql_type=connect)
        self.log.info(sql_res8)
        self.assertIn('REVOKE', sql_res8, '执行失败' + text)
        self.assertIn('cmk_show', sql_res8, '执行失败' + text)
        self.assertIn('SET', sql_res3, '执行失败' + text)
        self.assertIn('(3 rows)', sql_res8, '执行失败' + text)

    def tearDown(self):
        text = '--step5:清理环境;expect:成功--'
        self.log.info(text)
        sql_cmd = f'drop column encryption key if exists cek_show cascade;' \
                  f'drop client master key if exists cmk_show cascade;' \
                  f'reset session authorization;' \
                  f'drop owned by u_sg_tester09 cascade;'
        self.log.info(sql_cmd)
        connect = f'-d {self.db_name} -U {self.db_user} -W {macro.COMMON_PASSWD} -C'
        sql_res = self.sh_primary.execut_db_sql(sql_cmd, sql_type=connect)
        self.log.info(sql_res)
        sql_clean = f'drop user u_sg_tester09;' \
                    f'set dolphin.b_compatibility_mode=off;' \
                    f"drop database if exists {self.db_name} ;"
        self.log.info(sql_clean)
        clean_res = self.sh_primary.execut_db_sql(sql_clean)
        self.log.info(clean_res)
        self.assertNotIn("ERROR", sql_res, '执行失败' + text)
        self.log.info("===删除localkms路径，except：成功===")
        mkdir_cmd = f'source {macro.DB_ENV_PATH};' \
                    f'rm -rf {self.tmp_path};' \
                    f'ls {self.tmp_path}'
        self.log.info(mkdir_cmd)
        mkdir_res = self.user_node.sh(mkdir_cmd).result()
        self.log.info(mkdir_res)
        self.assertIn("No such file or directory", mkdir_res, '执行失败' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
